package data.FlinkJoin

import java.lang
import java.sql.Timestamp

import org.apache.commons.lang3.time.FastDateFormat
import org.apache.flink.api.common.functions.{CoGroupFunction, JoinFunction, RichMapFunction}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.{OutputTag, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.scala.function.AllWindowFunction
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.JavaConverters._
/**
 * 思想：
 * items.coGroup(orders)
 * 1) items（左边） late ==> side + API/DB
 * 2) order（右边） late ==> API/DB
 * 3) 最后2边数据union
 *
 *JVM GC参数
 * -Xmx  -Xms 至少4G以上 2个参数一致
 * SparkHistoryServer: 16G
 * ThriftServer/HiveServer2 调大内存
 */
object JoinApp {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    test02(env)

    env.execute(this.getClass.getCanonicalName)
  }

  def test02(env:StreamExecutionEnvironment): Unit = {
    val window = 10
    val delay = 5002


    val late = new OutputTag[(String, String, Long)]("late")

    val left = env.addSource(new ASource).assignTimestampsAndWatermarks(new DataAssignerWithPeriodicWatermarks(delay))
    val right = env.addSource(new BSource).assignTimestampsAndWatermarks(new DataAssignerWithPeriodicWatermarks(delay))

    val leftStream = left.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
        .sideOutputLateData(late)
        .apply(new AllWindowFunction[(String, String, Long), (String, String, Long), TimeWindow] {
          override def apply(window: TimeWindow, input: Iterable[(String, String, Long)], out: Collector[(String, String, Long)]) = {
            for(ele <- input) {
              out.collect(ele)
            }
          }
        })

    // 处理延迟数据
    leftStream.getSideOutput(late).map(new RichMapFunction[(String, String, Long),String] {
      override def open(parameters: Configuration): Unit = super.open(parameters)

      override def close(): Unit = super.close()

      override def map(value: (String, String, Long)): String = {
        // itemid orderid  => order找
        ""
      }
    })

    leftStream.coGroup(right).where(_._1).equalTo(_._1)
      .window(TumblingEventTimeWindows.of(Time.seconds(5)))
      .apply(new CoGroupFunction[(String, String, Long), (String, String, Long), (String, String, String, Long, Long)] {
        override def coGroup(first: lang.Iterable[(String, String, Long)],
                              second: lang.Iterable[(String, String, Long)],
                              out: Collector[(String, String, String, Long, Long)]) = {

          for(firstEle <- first.asScala) {
            var flag = false
            for(secondEle <- second.asScala) {
              out.collect((firstEle._1, firstEle._2, secondEle._2, firstEle._3, secondEle._3))
              flag = true
            }
            if(!flag) {
              out.collect((firstEle._1, firstEle._2, "null", firstEle._3, -1L))
            }
          }
        }
      }).print()
  }

  /**
   * 前提条件：同窗口同key
   */
  def test01(env:StreamExecutionEnvironment): Unit = {
    val window = 10
    val delay = 5002  // 5001 不行 深圳数据不会join 因为115000-5001=109999 [100,110)窗口左闭右开 wm>end边界 已触发

    val left = env.addSource(new ASource).assignTimestampsAndWatermarks(new DataAssignerWithPeriodicWatermarks(delay))
    val right = env.addSource(new BSource).assignTimestampsAndWatermarks(new DataAssignerWithPeriodicWatermarks(delay))

    left.join(right).where(_._1).equalTo(_._1)
        .window(TumblingEventTimeWindows.of(Time.seconds(window)))
        .apply(new JoinFunction[(String, String, Long),(String, String, Long),(String,String,String,Long,Long)] {
          override def join(first: (String, String, Long), second: (String, String, Long)): (String, String, String, Long, Long) = {
            (first._1, first._2, second._2, first._3, second._3)
          }
        }).print()
  }
}


class DataAssignerWithPeriodicWatermarks(maxAllowedUnOrderedTime: Long) extends AssignerWithPeriodicWatermarks[(String,String,Long)] {

  val format = FastDateFormat.getInstance("HH:mm:ss")

  var currentMaxTimestamp: Long = _

  override def getCurrentWatermark: Watermark = new Watermark(currentMaxTimestamp - maxAllowedUnOrderedTime)

  override def extractTimestamp(element: (String,String,Long), recordTimestamp: Long): Long = {
    val nowTime = element._3
    currentMaxTimestamp = currentMaxTimestamp.max(nowTime)

//    println(s"${element} -- ${nowTime} -- ${format.format(nowTime)} -- ${format.format(currentMaxTimestamp)} -- ${format.format(getCurrentWatermark.getTimestamp)}")

    nowTime
  }
}